package org.example.flink.stream.demo;

import java.util.ArrayList;
import java.util.List;

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.shaded.guava18.com.google.common.collect.Lists;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

public class UnionDemo {

    public void run() throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        List<String> dataSource = Lists.newArrayList();
        dataSource.add("a");
        dataSource.add("b");
        dataSource.add("c");
        DataStreamSource<String> source = env.fromCollection(dataSource);

        List<String> dataSource2 = Lists.newArrayList();
        dataSource2.add("e");
        dataSource2.add("f");
        dataSource2.add("g");
        DataStreamSource<String> source2 = env.fromCollection(dataSource2);

        DataStream<String> unionStream = source.union(source2).map(key -> {
            return key + "=>OK";
        });
        unionStream.print();
        env.execute("FromCollectionStream");
    }

    public static void main(String[] args) {
        try {
            new UnionDemo().run();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}
